package com.cjfather.redis.steam.annotation.config;

import java.lang.reflect.Method;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.Vector;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Component;

import com.cjfather.redis.steam.annotation.AnnoRedisStreamListener;
import com.cjfather.redis.steam.annotation.consumers.AnnoObjeStringConsumerI;
import com.cjfather.redis.steam.handler.CustomErrorHandler;
import com.cjfather.redis.steam.listener.StreamConsumerListener;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author: todd
 * @date:   2023-08-08
 */

@Slf4j
@Configuration
@Component
@Order(2)
public class RedisObjStringStreamConfig  implements DisposableBean {


	@Resource
	private ApplicationContext context;

	@Resource
	private StringRedisTemplate stringRedisTemplate;

	private Vector<StreamMessageListenerContainer<String, ObjectRecord<String, String>>> containerList = new Vector<>();

	@Resource
	RedisConnectionFactory factory;
	
	@Resource
	StreamConsumerListener streamConsumerListener;

	@PostConstruct
	public void initRedisStream() throws Exception {
//		Map<String, Object> beansWithAnnotation = context.getBeansWithAnnotation(AnnoRedisStreamListener.class);
		Map<String, AnnoObjeStringConsumerI> beansWithAnnotation = context.getBeansOfType(AnnoObjeStringConsumerI.class);
		if (beansWithAnnotation.size() == 0) {
			return;
		}
		
		for (Object item : beansWithAnnotation.values()) {
			if (!(item instanceof StreamListener)) {
				continue;
			}
			//Record取不到值
			Method method = item.getClass().getDeclaredMethod("onMessage", ObjectRecord.class);
			AnnoRedisStreamListener annotation = method.getAnnotation(AnnoRedisStreamListener.class);
			if (annotation == null) {
				log.info("queue is null");
				continue;
			}
			creasteSubscription( (AnnoObjeStringConsumerI) item, annotation.streamKey(), annotation.consumerGroup(), annotation.consumerName(), annotation.consumerSize());
			
//			Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(item.getClass(),
//					(ReflectionUtils.MethodFilter) method2 ->
//							AnnotationUtils.findAnnotation(method2, AnnoRedisStreamListener.class) != null);
//			
//			if (methodsWithHandler == null || methodsWithHandler.size() < 1) {
//				log.info("queue is null");
//				continue;
//			}
//			for(Method me:methodsWithHandler) {
//				System.out.println(me.getName());
//				AnnoRedisStreamListener annotation = me.getAnnotation(AnnoRedisStreamListener.class);
//				creasteSubscription(factory, (AnnoObjectConsumerI) item, annotation.streamKey(), annotation.consumerGroup(), annotation.consumerName(), annotation.consumerSize());
//			}
		}
	}
	
	private Set<AnnoRedisStreamListener> findListenerAnnotations(Method method) {
		Set<AnnoRedisStreamListener> listeners = new HashSet<>();
		AnnoRedisStreamListener ann = AnnotatedElementUtils.findMergedAnnotation(method, AnnoRedisStreamListener.class);
		if (ann != null) {
			listeners.add(ann);
		}
		
		return listeners;
	}

	private void creatGroup(String key, String group) {
		StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();
		try {
			streamOperations.createGroup(key,ReadOffset.from("0-0"), group);
		} catch (Exception e) {
			log.info("{}组存在",group);
		}
		log.info("creatGroup:{}", group);
	}

	private void deleteGroup(String key, String group) {
		StreamOperations<String, String, Map> streamOperations = this.stringRedisTemplate.opsForStream();
		try {
			streamOperations.destroyGroup(key, group);
		} catch (Exception e) {
			log.error("{}组delete",group,e);
		}
		log.info("deleteGroup:{}", group);
	}
	
	public void creasteSubscription(AnnoObjeStringConsumerI streamListener, String streamKey, String group, String consumerName,int consumerSize) {
		StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();
		if (stringRedisTemplate.hasKey(streamKey)) {
			StreamInfo.XInfoGroups groups = streamOperations.groups(streamKey);
			if (groups.isEmpty()) {
				creatGroup(streamKey, group);
			} else {
				groups.stream().forEach(g -> {
					log.info("XInfoGroups:{}", g);
					StreamInfo.XInfoConsumers consumers = streamOperations.consumers(streamKey, g.groupName());
					this.deleteGroup(streamKey, group);
					creatGroup(streamKey, group);
					log.info("XInfoConsumers:{}", consumers);
				});
			}
		} else {
			creatGroup(streamKey, group);
		}

		StreamMessageListenerContainer<String, ObjectRecord<String, String>> listenerContainer = streamConsumerListener.getContainer2(streamKey);
		
		for(int i = 0;i < consumerSize;i++) {
			StreamOffset<String> streamOffset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());
			Consumer consumer = Consumer.from(group, consumerName + "-" + i);
			log.info("队列{} 组{} 消费者{}", streamKey , group, consumer.getName());
			StreamMessageListenerContainer.ConsumerStreamReadRequest<String> streamReadRequest = StreamMessageListenerContainer
	                .StreamReadRequest
	                .builder(streamOffset)
	                .consumer(consumer)
	                // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
	                .errorHandler(new CustomErrorHandler(stringRedisTemplate, streamKey, group,streamListener.getClass().getName()))
	                .autoAcknowledge(true)
	                // 如果消费者发生了异常，判断是否取消消费者消费
	                .cancelOnError(throwable -> false)
	                .build();
			
			AnnoObjeStringConsumerI bean = context.getBean(streamListener.getClass());
			bean.setGroup(group);
			bean.setConsumerName(consumer.getName());
			listenerContainer.register(streamReadRequest, bean);
		}
		
		this.containerList.add(listenerContainer);
		listenerContainer.start();
	}
	


	@Override
	public void destroy() {
		this.containerList.forEach(StreamMessageListenerContainer::stop);
	}
}
